using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Ddl;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Locking;

namespace Me.Prettyprint.Cassandra.Locking
{
    /// <summary>
    /// Base class for lock managers. The methods that subclasses need to implement are:
    /// Acquire, Release and CreateLock.
    /// </summary>
    /// <typeparam name="K"></typeparam>
	
	public abstract class AbstractLockManager<K> : HLockManager
	{
        protected const string DUMMY_VALUE = "v";
        private static Logger logger = new Logger(typeof(AbstractLockManager<K>));

		public AbstractLockManager(Cluster cluster, Keyspace keyspace, HLockManagerConfigurator lockManagerConfigurator)
		{
			if (cluster == null)
            {
				throw new System.SystemException("Cluster cannot be null for LockManager");
            }
			
			Cluster = cluster;
			LockManagerConfigurator = lockManagerConfigurator;
			if (lockManagerConfigurator == null)
			{
				LockManagerConfigurator = new HLockManagerConfigurator();
			}
			
			if (Keyspace == null)
			{
				Keyspace = HFactory.CreateKeyspace(LockManagerConfigurator.KeyspaceName, cluster);
			}
			else
			{
				Keyspace = keyspace;
				// Set the Keyspace name in order to keep the info consistent
				LockManagerConfigurator.KeyspaceName = keyspace.KeyspaceName;
			}
		}
		public AbstractLockManager(Cluster cluster):this(cluster, null, null)
		{
		}
		
		public AbstractLockManager(Cluster cluster, HLockManagerConfigurator hlmc):this(cluster, null, hlmc)
		{
		}
		
		public AbstractLockManager(Cluster cluster, Keyspace keyspace):this(cluster, keyspace, null)
		{
		}
        public Cluster Cluster
        {
            get;
            set;
        }
        public Keyspace Keyspace
        {
            get;
            set;
        }
        public HLockManagerConfigurator LockManagerConfigurator
        {
            get;
            set;
        }
		public CassandraHostConfigurator Configurator
		{
			get
			{
				return ((AbstractCluster) Cluster).Configurator;
			}
            set
            {
                ((AbstractCluster)Cluster).Configurator = value;
            }
		}
		public virtual void Init()
		{
			CheckCreateLockSchema();
		}
		private void CheckCreateLockSchema()
		{
			KeyspaceDefinition keyspaceDef = Cluster.DescribeKeyspace(Keyspace.KeyspaceName);
			if (keyspaceDef == null)
			{
				ColumnFamilyDefinition cfDef = CreateColumnFamilyDefinition();
                List<ColumnFamilyDefinition> cfdefs = new List<ColumnFamilyDefinition>();
                cfdefs.Add(cfDef);
				KeyspaceDefinition newKeyspace = HFactory.CreateKeyspaceDefinition(Keyspace.KeyspaceName, ThriftKsDef.DEF_STRATEGY_CLASS,
                    LockManagerConfigurator.ReplicationFactor, cfdefs);
				Cluster.AddKeyspace(newKeyspace, true);
			}
			else
			{
				if (!DoesLockCFExist(keyspaceDef))
				{
					// create it
					ColumnFamilyDefinition cfDef = CreateColumnFamilyDefinition();
					Cluster.AddColumnFamily(cfDef, true);
				}
				else
				{
				}
			}
		}
		
		private ColumnFamilyDefinition CreateColumnFamilyDefinition()
		{
			ColumnFamilyDefinition cfDef = HFactory.CreateColumnFamilyDefinition(Keyspace.KeyspaceName, LockManagerConfigurator.LockManagerCF, ComparatorType.UTF8TYPE);
			cfDef.KeyValidationClass = ComparatorType.UTF8TYPE.ClassName;
			cfDef.RowCacheSize = LockManagerConfigurator.RowsCacheEnabled?10000:0;
			return cfDef;
		}
		
		private bool DoesLockCFExist(KeyspaceDefinition keyspaceDef)
		{
			foreach (ColumnFamilyDefinition cfdef in keyspaceDef.CfDefs)
			{
				if (cfdef.Name.Equals(LockManagerConfigurator.LockManagerCF))
				{
					return true;
				}
			}
			return false;
		}
        public abstract void Acquire(HLock hlock, long timeout);
        public abstract void Acquire(HLock hlock);
		public abstract void Release(HLock hlock);
        public abstract HLock CreateLock(string value);
    }
}